package com.hzlj.position.job.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.fz.common.base.core.JsonUtils;
import com.fz.common.base.exception.PubException;
import com.hzlj.position.job.common.model.NoticeJob;
import com.hzlj.position.config.config.PositionConfig;
import com.hzlj.position.job.dao.NoticeJobDao;
import com.hzlj.position.job.notice.BaseNoticeJob;
import com.hzlj.position.job.service.NoticeJobService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.support.CronSequenceGenerator;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 通知任务配置(NoticeJob)表服务接口
 *
 * @author makejava
 * @date 2024-09-09 11:52:37
 */
@Slf4j
@Service
public class NoticeJobServiceImpl implements NoticeJobService {
    @Resource
    private NoticeJobDao noticeJobDao;
    @Resource
    private PositionConfig positionConfig;
    @Resource
    private ApplicationContext applicationContext;
    private BlockingQueue<List<NoticeJob>> schedulesQueue;

    @PostConstruct
    public void init() {
        schedulesQueue = new ArrayBlockingQueue<>(positionConfig.getNoticeJobQueue().getSize());
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(positionConfig.getNoticeJobQueue().getThread());

        executor.scheduleWithFixedDelay(() -> {
            try {
                List<NoticeJob> jobs = schedulesQueue.take();
                schedule(jobs);
            } catch (Exception e) {
                log.error("【通知JOB】取出调度QUEUE失败", e);
            }
        }, 0, positionConfig.getNoticeJobQueue().getFixedDelaySecond(), TimeUnit.SECONDS);
    }

    @Override
    public void schedule() {
        List<NoticeJob> noticeJobs = noticeJobDao.listNoticeJob();
        if (ObjectUtil.isEmpty(noticeJobs)) {
            log.warn("【通知JOB】没有需要调度的任务");
            return;
        }

        //修改状态
        noticeJobDao.beginSchedule(noticeJobs.stream().map(NoticeJob::getId).collect(Collectors.toList()));

        Map<String, List<NoticeJob>> noticeJobsByHandler = noticeJobs.stream().collect(Collectors.groupingBy(NoticeJob::getHandler));
        noticeJobsByHandler.forEach((k, v) -> {
            try {
                schedulesQueue.put(v);
            } catch (Exception e) {
                log.error("【通知JOB】写入调度QUEUE失败:jobs={}", JsonUtils.obj2json(v), e);
            }
        });
    }


    private void schedule(List<NoticeJob> jobs) {
        boolean success = false;
        try {
            log.info("【通知JOB】统计开始:jobs={}", jobs);
            exe(jobs);
            success = true;
        } catch (Exception e) {
            log.info("【通知JOB】统计异常:jobs={}", jobs, e);
        } finally {
            for (NoticeJob job : jobs) {
                Date nextExeTime = job.getNextExeTime();
                if (success) {
                    nextExeTime = this.nextExeTime(job);
                }
                log.info("【通知JOB】统计结束:success={},id={},jgCode={},handler={},nextExeTime={}",
                        success, job.getId(), job.getJgCode(), job.getHandler(), nextExeTime);
                this.noticeJobDao.endSchedule(job.getId(), nextExeTime);
            }
        }
    }


    private void exe(List<NoticeJob> jobs) {
        NoticeJob job = jobs.get(0);
        try {
            BaseNoticeJob handler = applicationContext.getBean(StrUtil.lowerFirst(job.getHandler()), BaseNoticeJob.class);
            //调用
            handler.notice(jobs);
        } catch (Exception e) {
            log.error("【通知JOB】通知失败:handler={},jobs={}",
                    job.getHandler(), JsonUtils.obj2json(jobs), e);
        }
    }


    private Date nextExeTime(NoticeJob noticeJob) {
        Date now = new Date();
        //如果比当前时间还大
        if (noticeJob.getNextExeTime().compareTo(now) > 0) {
            return noticeJob.getNextExeTime();
        }

        String scheduleCron = noticeJob.getScheduleCron();
        if (ObjectUtil.isEmpty(scheduleCron)) {
            throw new PubException("请确认是否配置了调度表达式");
        }

        CronSequenceGenerator cronSequenceGenerator = new CronSequenceGenerator(scheduleCron);

        log.info("【通知JOB】设置下一次调度时间:id={},cron={},nextTime={}",
                noticeJob.getId(), scheduleCron, DateUtil.formatDateTime(cronSequenceGenerator.next(now)));
        return cronSequenceGenerator.next(now);
    }

}
